package com.dd.common.util

import android.text.TextUtils
import com.dd.common.base.BaseResponse
import com.dd.common.base.CommonCallback
import com.dd.lnetwork.exception.ExceptionNetWork
import com.dd.lnetwork.exception.ServerException
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.core.ObservableSource
import io.reactivex.rxjava3.core.ObservableTransformer
import io.reactivex.rxjava3.core.SingleTransformer
import io.reactivex.rxjava3.disposables.Disposable
import io.reactivex.rxjava3.functions.Function
import io.reactivex.rxjava3.schedulers.Schedulers
import org.reactivestreams.Publisher
import java.util.concurrent.TimeUnit

object RxUtil {

    /**
     * 统一线程处理
     * compose简化线程
     */
    fun <T : Any> rxSchedulerHelper(): ObservableTransformer<T, T> {
        return ObservableTransformer { observable ->
            observable.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
        }
    }

    /**
     * 统一线程处理
     */
    fun <T : Any> rxSingleSchedulerHelper(): SingleTransformer<T, T> {    //compose简化线程
        return SingleTransformer { upstream ->
            upstream.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
        }
    }

    private class HttpResponseFunc<T : Any> : Function<Throwable, Observable<T>> {
        override fun apply(throwable: Throwable): Observable<T> {
            return Observable.error(ExceptionNetWork.handleException(throwable))
        }
    }

    /**
     * 统一返回结果处理
     */
    fun <T : Any> handleResult(): ObservableTransformer<BaseResponse<T>, T> {
        return ObservableTransformer { upstream ->
            upstream.flatMap { response ->
                if (!TextUtils.equals("000000", response.code)) {
                    Observable.error(ServerException(response.code, response.msg))
                } else createData(response.data)
            }.onErrorResumeNext(HttpResponseFunc())
        }
    }

    /**
     * 统一返回结果处理
     */
    fun <T : Any> handleResult(errorFunction: CommonCallback<String>): ObservableTransformer<BaseResponse<T>, T> {
        return ObservableTransformer { upstream ->
            upstream.flatMap { response ->
                if (!TextUtils.equals("000000", response.code)) {
                    errorFunction.call(response.msg)
                    Observable.error(ServerException(response.code, response.msg))
                } else createData(response.data)
            }.onErrorResumeNext(HttpResponseFunc())
        }
    }

    /**
     * 生成Observable
     */
    private fun <T : Any> createData(t: T): ObservableSource<T> {
        return ObservableSource { observer ->
            try {
                observer.onNext(t)
                observer.onComplete()
            } catch (e: Exception) {
                observer.onError(e)
            }
        }
    }

    /**
     * 生成Observable
     *
     * @param <T>
     * @return
    </T> */
    fun <T> createPublisherData(t: T): Publisher<T> {
        return Publisher { observer ->
            try {
                observer.onNext(t)
                observer.onComplete()
            } catch (e: Exception) {
                observer.onError(e)
            }
        }
    }

    /**
     * 等待几毫秒执行任务,返回Subscription，为解决多次重复触发，可调用unsubscribe()取消前一次定时器
     *
     * @param millisecond
     * @param action
     */
    fun timer(millisecond: Long, action: CommonCallback<Long>): Disposable {
        return Observable.timer(millisecond, TimeUnit.MILLISECONDS)
            .compose(rxSchedulerHelper())
            .subscribe({ aLong -> action.call(aLong) }
            ) { throwable -> throwable.printStackTrace() }
    }

    fun interval(times: Long, action: CommonCallback<Long>): Disposable {
        return Observable.interval(0, 1000, TimeUnit.MILLISECONDS).take(times)
            .compose(rxSchedulerHelper())
            .subscribe({ l: Long ->
                action.call(l)
            }) { throwable: Throwable -> throwable.printStackTrace() }
    }
}